package com.yh.csx.bsf.ods.fullload;

import java.io.File;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.Map.Entry;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.Header;
import com.yh.csx.bsf.core.base.Ref;
import com.yh.csx.bsf.core.db.DbConn;
import com.yh.csx.bsf.core.util.ContextUtils;
import com.yh.csx.bsf.core.util.ConvertUtils;
import com.yh.csx.bsf.core.util.FileUtils;
import com.yh.csx.bsf.core.util.LogUtils;
import com.yh.csx.bsf.core.util.PropertyUtils;
import com.yh.csx.bsf.ods.meta.OdsSyncAdapter;
import com.yh.csx.bsf.ods.AbstractCanalConsumer;
import com.yh.csx.bsf.ods.StragtegyProvider;
import com.yh.csx.bsf.ods.meta.OdsColumn;
import com.yh.csx.bsf.ods.meta.OdsColumn.OdsColumnBuilder;
import com.yh.csx.bsf.ods.meta.OdsIndex;
import com.yh.csx.bsf.ods.meta.OdsIndex.OdsIndexBuilder;
import com.yh.csx.bsf.ods.utils.WarnUtils;
import com.yh.csx.bsf.ods.meta.OdsTable;

import lombok.Getter;
import lombok.Setter;
import lombok.var;

/**
 * 全量消费者
 */

public class CanalFullLoadConsumer extends AbstractCanalConsumer {
	private AtomicInteger delayCount=new AtomicInteger(0);
	@Getter
	@Setter
	private TaskManager.SubTask task;
	@Getter
	@Setter
	private StragtegyProvider stragtegyProvider;
	@Getter
	@Setter
	private Map<String, OdsSyncAdapter> syncers;
	volatile boolean isFullloadOver = false;
	@Getter
	@Setter
	volatile boolean isIncreasingOver = false;

	@Override
	public void start() {
		super.start();
	}

	@Override
	public synchronized void stop() {
		this.isIncreasingOver = true;
		super.stop();
		LogUtils.info(this.getClass(), "ods", "【全量同步】增量部分" + gettable() + "结束。canal-cliet状态：" + super.isRunning());

	}

	@Override
	public void rename(CanalEntry.Header header, String sql) {
		super.rename(header, sql);
		try {
			// fullLoad();
		} catch (Throwable e) {
			close(e);
		}

	}

	public void renameTempToOrginal() {
		// 删除原表并重命名fullload表(后续考虑分布式锁)
		// mysql 表重命名 to
		if (this.getTask().getParent().isSucceed()) {
			com.yh.csx.bsf.redis.RedisProvider redisProvider = null;
			try {
				redisProvider = ContextUtils.getBean(com.yh.csx.bsf.redis.RedisProvider.class, true);
			} catch (Exception e) {
			}
			if (redisProvider != null) {
				redisProvider.lock(task.getTargetTable(), 160, () -> {
					try {//暂停2s等待实时增量同步完
						Thread.sleep(2000);
					}catch(Exception e) {}
					for (OdsSyncAdapter syncer : syncers.values()) {
						try {
							syncer.dropTable(task.getTargetTable().trim());
							syncer.renameTable(task.getTargetTable() + "_temp", task.getTargetTable());
						} catch (Exception e) {
							task.setFail();
							LogUtils.error(this.getClass(), "ods", "【全量同步:" + syncer.getClass() + "】" + gettable() + "结束", e);
						}
					}
				});
			} else {
				for (OdsSyncAdapter syncer : syncers.values()) {
					try {
						syncer.dropTable(task.getTargetTable());
						syncer.renameTable(task.getTargetTable() + "_temp", task.getTargetTable());
					} catch (Exception e) {
						task.setFail();
						LogUtils.error(this.getClass(), "ods", "【全量同步:" + syncer.getClass() + "】" + gettable() + "结束",e);
					}
				}
			}
			LogUtils.info(this.getClass(), "ods", "【全量同步】" + gettable() + "结束");
		}

	}

	@Override
	public void connectorBatchArk(long batchId) throws Exception {
		/*
		 * try { super.connectorBatchArk(batchId);
		 *//**
			 * 执行最新更新
			 *//*
				 * Optional.of(syncers).get().forEach((name, syncer) -> { try {
				 * syncer.process(batchOperate); } catch (Exception e) { } });
				 * 
				 * // 检查是否结束更新 if (this.batchOperate.size() == 0) {
				 * checkEnd(System.currentTimeMillis()); } else {
				 * checkEnd(this.batchOperate.get(this.batchOperate.size() -
				 * 1).getHeader().getExecuteTime()); } batchOperate.clear(); } catch (Exception
				 * e) { close(e); }
				 */
	}

	private void checkEnd(long executeTime) {
		long delay = System.currentTimeMillis() - executeTime;
		LogUtils.info(this.getClass(), "dds", "【全量同步】" + gettable() + "延迟" + delay + "ms");
		// 最近2秒内有同步,则结束
		if (delay < PropertyUtils.getPropertyCache("canal.ods.delay.limit", 1000)) {
			delayCount.getAndIncrement();
			if(delayCount.getAndIncrement()>1000) {
				close(null);
			}
		}
	}

	private String gettable() {
		return task.getDb() + "." + task.getTable();
	}

	public void fullLoad() throws Throwable {
		task.setProcess(0);
		LogUtils.info(this.getClass(), "dds", "【全量同步】" + gettable() + "开始");
		// 新建table_fullload表 如果不存在
		// mysql 结构to clickhouse
		OdsTable tableStructure = new OdsTable(task.getDb(), task.getTable());
		try (DbConn dbConn = stragtegyProvider.getDb(task.getDb())) {
			if (dbConn == null)
				throw new RuntimeException("未配置数据库数据源:" + task.getDb());
			ResultSet rs = dbConn.executeResultSet(
					"select * from information_schema.columns where table_schema =? and table_name=?",
					new Object[] { task.getDb(), task.getTable() });
			List<OdsColumn> columns = new ArrayList<OdsColumn>();
			// 同步表结构
			boolean idExists = false;
			while (rs.next()) {
				columns.add(OdsColumnBuilder.build().setColumnName(rs.getString("COLUMN_NAME"))
						.setOrdinalPosition(Integer.valueOf(rs.getString("ORDINAL_POSITION").replaceAll(",", "")))
						.setColumnDefault(rs.getString("COLUMN_DEFAULT"))
						//.setNullable("YES".equals(rs.getString("IS_NULLABLE").toUpperCase()) ? true : false)
						.setDataType(rs.getString("DATA_TYPE"))
						.setCharacterMaxmunLength(Long.valueOf(Optional
								.ofNullable(rs.getString("CHARACTER_MAXIMUM_LENGTH")).orElse("0").replaceAll(",", "")))
						.setCharacterOctetLength(Long.valueOf(Optional
								.ofNullable(rs.getString("CHARACTER_OCTET_LENGTH")).orElse("0").replaceAll(",", "")))
						.setNumericPrecision(rs.getInt("NUMERIC_PRECISION")).setNumericScale(rs.getInt("NUMERIC_SCALE"))
						.setDatetimePrecision(rs.getInt("DATETIME_PRECISION"))
						.setCharacterSetName(rs.getString("CHARACTER_SET_NAME"))
						.setCollationName(rs.getString("COLLATION_NAME")).setColumnType(rs.getString("COLUMN_TYPE"))
						.setColumnKey(rs.getString("COLUMN_KEY")).setExtra(rs.getString("EXTRA"))
						.setComment(rs.getString("COLUMN_COMMENT")));
				if ("ID".equals(rs.getString("COLUMN_NAME").toUpperCase())) {
					idExists = true;
				}
			}
			rs.close();
			if (!idExists) {
				task.setFail();
				throw new Exception("【全量同步】" + gettable() + "同步id字段不存在");
			}
			rs = dbConn.executeResultSet("show index from " + task.getDb() + "." + task.getTable(), null);
			List<OdsIndex> indexs = new ArrayList<OdsIndex>();
			while (rs.next()) {
				indexs.add(OdsIndexBuilder.build().setColumnName(rs.getString("Column_name"))
						.setKeyName(rs.getString("Key_name")).setCollation(rs.getString("Collation"))
						.setSeqInIndex(rs.getInt("Seq_in_index")).setComment(rs.getString("Comment"))
						.setIndexType(rs.getString("Index_type")).setNonUnique(rs.getInt("Non_unique")));
			}
			rs.close();
			tableStructure.setIndexs(indexs);
			tableStructure.setColumns(columns);
			syncers.forEach((name, syncer) -> {
				try {	
					Thread.sleep(1000);
					syncer.createTableIfNotExists(task.getTargetTable() + "_temp", columns);
					syncer.createIndex(task.getTargetTable() + "_temp", indexs);
				} catch (Exception e) {
					task.setFail();
					LogUtils.error(this.getClass(), "ods", "【全量同步】" + gettable() + "同步数据结构异常", e);
					return;

				}
			});
			// 循环批量导入到fullload表
			final Ref<Long> maxId = new Ref(Long.valueOf(-1L));
			long count = 0;
			long allCount = ConvertUtils.convert(dbConn.executeScalar("select count(0) from {db}.{table}"
					.replace("{db}", task.getDb()).replace("{table}", task.getTable()), null), long.class);

			if (allCount == 0) {// 如果数据为空，则返回
				task.setProcess(1);
				return;
			}
			while (count < allCount && !isFullloadOver) {
				List<Map<String, Object>> data = dbConn.executeList(
						"select * from {db}.{table} where id>? order by id asc limit {maxcount}"
								.replace("{db}", task.getDb()).replace("{table}", task.getTable())
								.replace("{maxcount}", PropertyUtils.getPropertyCache("fullload.maxcount", 200) + ""),
						new Object[] { maxId.getData() });

				if (data.size() == 0) {
					task.setSuccess();
					isFullloadOver = true;
					return;
				} else {
					syncers.forEach((name, syncer) -> {
						try {
							syncer.insertBatchMap(task.getTargetTable() + "_temp", data, tableStructure);
						} catch (Throwable e) {
							task.setFail();
							isFullloadOver = true;
							LogUtils.error(this.getClass(), "ods", "【全量同步】" + gettable() + "同步数据", e);
							return;
						}
					});
					count += data.size();
					if (allCount > 0) {
						task.setProcess(((double) count / (double) allCount));
						LogUtils.info(this.getClass(), "ods","【全量同步】" + gettable() + "同步数据进度：" + allCount + "," + count);
					} else {
						task.setProcess(1);
					}

				}
				data.forEach(c -> {
					Long id = Long.valueOf(c.get("id") == null ? "0" : c.get("id").toString());
					if (id.longValue() > maxId.getData().longValue()) {
						maxId.setData(id);
					}
				});
			}
		} catch (Exception e) {
			task.setFail();
			LogUtils.error(this.getClass(), "ods", "【全量同步】" + gettable() + "同步异常：", e);
		}
		isFullloadOver = true;
	}

	protected void close(Throwable e) {
		if (e == null) {
			task.setSuccess();
			LogUtils.info(this.getClass(), "ods", "【消费者】" + gettable() + "发起关闭操作");
		} else {
			task.setFail();
			LogUtils.error(this.getClass(), "ods", "【消费者】" + gettable() + "异常,发起关闭操作", e);
		}
		this.stop();
		/**
		 * 清理canal同步配置
		 */
		removeProperty(task);

	}

	@Override
	public void insert(Header header, List<Column> afterColumns) throws Exception {
		Optional.of(syncers).ifPresent(c -> {
			for (OdsSyncAdapter syncer : c.values()) {
				try {
					syncer.insert(header, afterColumns);
				} catch (Exception e) {
					task.setFail();
					LogUtils.error(this.getClass(), "dds", "【全量同步】异常" + gettable() + "结束", e);
				}
			}
		});
	}

	@Override
	public void update(Header header, List<Column> beforeColumns, List<Column> afterColumns) throws Exception {
		Optional.of(syncers).ifPresent(c -> {
			for (OdsSyncAdapter syncer : c.values()) {
				try {
					syncer.insert(header, afterColumns);
				} catch (Exception e) {
					task.setFail();
					LogUtils.error(this.getClass(), "dds", "【全量同步】异常" + gettable() + "结束", e);
				}
			}
		});

	}

	@Override
	public void delete(Header header, List<Column> beforeColumns) throws Exception {
		Optional.of(syncers).ifPresent(c -> {
			for (OdsSyncAdapter syncer : c.values()) {
				try {
					syncer.delete(header, beforeColumns);
				} catch (Exception e) {
					task.setFail();
					LogUtils.error(this.getClass(), "dds", "【全量同步】异常" + gettable() + "结束", e);
				}
			}
		});
	}

	@Override
	public boolean insertBatch(com.alibaba.otter.canal.protocol.CanalEntry.Entry[] entries) throws Exception {
		String database=entries[0].getHeader().getSchemaName();
		String table=entries[0].getHeader().getTableName();
		String targetTable=stragtegyProvider.getTargetTable(database, table);		
		for (Entry<String, OdsSyncAdapter> syncer : syncers.entrySet()) {
			try {
				syncer.getValue().insertBatch(targetTable,entries); 
			} catch (Exception e) {
				LogUtils.error(this.getClass(), "dds", "【增量同步】异常" +targetTable + "", e);
				WarnUtils.alarm("增量同步异常",database+"."+table+":update:", e);
				return false;
			}
		}
		return true;
	}

	@Override
	public boolean updateBatch(com.alibaba.otter.canal.protocol.CanalEntry.Entry[] entries) throws Exception {
		String database=entries[0].getHeader().getSchemaName();
		String table=entries[0].getHeader().getTableName();
		String targetTable=stragtegyProvider.getTargetTable(database, table);		
		for (Entry<String, OdsSyncAdapter> syncer : syncers.entrySet()) {
			try {
				syncer.getValue().updateBatch(targetTable,entries); 
			} catch (Exception e) {
				LogUtils.error(this.getClass(), "dds", "【增量同步】异常" +targetTable + "", e);
				WarnUtils.alarm("增量同步异常",database+"."+table+":update:", e);
				return false;
			}
			
		}
		return true;
	}

	@Override
	public boolean deleteBatch(com.alibaba.otter.canal.protocol.CanalEntry.Entry[] entries) throws Exception {
		String database=entries[0].getHeader().getSchemaName();
		String table=entries[0].getHeader().getTableName();
		String targetTable=stragtegyProvider.getTargetTable(database, table);		
		for (Entry<String, OdsSyncAdapter> syncer : syncers.entrySet()) {
			try {
				syncer.getValue().deleteBatch(targetTable,entries); 
			} catch (Exception e) {
				LogUtils.error(this.getClass(), "dds", "【增量同步】异常" +targetTable + "", e);
				WarnUtils.alarm("增量同步异常",database+"."+table+":update:", e);
				return false;
			}
		}
		return true;
	}
	@Override
	public void whenOthers(CanalEntry.Entry entry) {
		checkEnd(entry.getHeader().getExecuteTime());
	}

	/**
	 * 消息处理完后，后处理
	 */
	@Override
	protected void afterProcessed(CanalEntry.Entry entry) {
		checkEnd(entry.getHeader().getExecuteTime());
	}

	/**
	 * 移除canal同步配置
	 */
	private void removeProperty(TaskManager.SubTask task) {
		var destination = task.getDb() + "_" + task.getTable();
		FileUtils.deleteDir(new File("canal/conf/" + destination));
	}
	
	@Override
	protected boolean processMessages(Message message) throws Exception {
		Map<String, List<CanalEntry.Entry>> mappingList = new HashMap<String, List<CanalEntry.Entry>>();
		for (CanalEntry.Entry entry : message.getEntries()) {
			if (entry.getEntryType().equals(EntryType.ROWDATA)) {
				String targetTable=stragtegyProvider.getTargetTable(entry.getHeader().getSchemaName() , entry.getHeader().getTableName());
				if (mappingList.containsKey(targetTable)) {
					mappingList.get(targetTable)
							.add(entry);
				} else {
					List<CanalEntry.Entry> list = new ArrayList<CanalEntry.Entry>();
					list.add(entry);
					mappingList.put(targetTable, list);
				}
			} else {
				whenOthers(entry);
			}
		}
		return processMessages0(mappingList);

	}	
}
